package mq

import (
	"container/list"
	"errors"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	log "ac-common-go/glog"
	"github.com/streadway/amqp"
)

const (
	IDLECHECKINTERVAL        = 60 //清除超时连接间隔
	REMOVEDPOOLCHECKINTERVAL = 60 //移除后池对象检测间隔
	DEFAULTCHANNEL           = 10 //一个conn对应的channel数
	EXCHANGETYPE             = "direct"
)

var nowFunc = time.Now

type NewChannel func(connection *amqp.Connection) (*amqp.Channel, error)

type PoolManager struct {
	poolList       list.List
	retryNum       int
	maxFailCount   int64
	lock           *sync.Mutex
	backupPoolList list.List
}

type Pool struct {
	lock        *sync.Mutex
	idle        list.List
	idleTimeout time.Duration
	connTimeout time.Duration
	maxConn     int
	initConn    int
	count       int
	url         string
	exchange    string
	routingKey  string
	failNum     int64
}

type IdleClient struct {
	connection *amqp.Connection
	channel    *amqp.Channel
	createTime time.Time
	destroy    bool
}

//error
var (
	ErrOverMax          = errors.New("over max connect num")
	ErrInvalidConn      = errors.New("client is nil when close")
	ErrPoolClosed       = errors.New("pool is already closed")
	ErrSocketDisconnect = errors.New("socket is already closed")
	ErrInvalidParam     = errors.New("param is invalid")
	ErrInvalidResult    = errors.New("result is invalid")
	ErrNoAvaliblePool   = errors.New("no avalible pool")
	ErrPoolManagerErr   = errors.New("pool manager is not avalible")
)

func NewPoolManager(urls string, maxConn, initConn, idleTimeout int,
	retryNum, maxFailCount int, exchange, routingkey string) *PoolManager {
	pm := &PoolManager{retryNum: retryNum, maxFailCount: int64(maxFailCount), lock: new(sync.Mutex)}
	urlArray := strings.Split(urls, "*")
	for idx := 0; idx < len(urlArray); idx++ {
		url := urlArray[idx]
		pool := &Pool{
			url:         url,
			lock:        new(sync.Mutex),
			maxConn:     maxConn,
			idleTimeout: time.Duration(idleTimeout) * time.Second,
			count:       0,
			initConn:    initConn,
			exchange:    exchange,
			routingKey:  routingkey,
		}
		if pm.poolList.Len() == 0 {
			err := pool.InitPool()
			if err == nil {
				pm.poolList.PushBack(pool)
			} else {
				log.Errorf("pool init error:%q", pool.GetPoolKey())
				pm.backupPoolList.PushBack(pool)
			}
		} else {
			pm.backupPoolList.PushBack(pool)
		}
	}

	go pm.StartIdleConnClearTask()
	return pm
}
func (pm *PoolManager) SendMessage(message string) error {
	if pm == nil {
		return ErrPoolManagerErr
	}
	var retErr error

	for i := 0; i <= pm.retryNum; i++ {
		poolEle := pm.GetPoolElement()
		if poolEle == nil {
			return ErrNoAvaliblePool
		}
		pool, ok := poolEle.Value.(*Pool)
		//xwj add begin 2019/1/17
		if !ok {
			log.Errorf("It's not ok for type Pool,poolEle.Value[%+v]", poolEle.Value)
			return errors.New("It's not ok for type Pool")
		}
		//xwj add end 2019/1/17
		retErr = pool.SendMessage(message)
		if retErr == nil {
			pool.resetFaileNum()
			break
		}
		if retErr == ErrInvalidParam || retErr == ErrOverMax {
			return retErr
		} else {
			pool.incrFaileNum()
			if pool.failNum >= pm.maxFailCount {
				log.Errorf("pool fail num is over max, will be removed:%q", pool.GetPoolKey())
				pm.RemovePoolElementToBackup(poolEle)
			}
			log.Errorf("idle client send rpc error:%+v, retry:%d", retErr, i)
		}
	}

	return retErr
}

func (pm *PoolManager) GetPoolElement() *list.Element {
	if pm.poolList.Len() < 1 {
		if pm.backupPoolList.Len() > 0 {
			pm.lock.Lock()
			if pm.poolList.Len() < 1 {
				for ele := pm.backupPoolList.Front(); ele != nil; ele = ele.Next() {
					backupPool, ok := ele.Value.(*Pool)
					//xwj add begin 2019/1/17
					if !ok {
						log.Errorf("It's not ok for type Pool,ele.Value[%+v]", ele.Value)
						break
					}
					//xwj add end 2019/1/17
					err := backupPool.InitPool()
					if err == nil {
						pm.poolList.PushBack(backupPool)
						break
					}
				}
			}
			pm.lock.Unlock()
		}
		return nil
	}
	poolEle := pm.poolList.Front()
	return poolEle
}

func (pm *PoolManager) RemovePoolElementToBackup(poolEle *list.Element) {
	pm.lock.Lock()
	pm.poolList.Remove(poolEle)
	pm.backupPoolList.PushBack(poolEle.Value)
	pm.lock.Unlock()
}

func (p *Pool) SendMessage(message string) error {
	idleClient, err := p.Get()
	if err != nil {
		log.Errorf("%s can not get idle client:%+v", p.GetPoolKey(), err)
		return err
	}
	defer p.Put(idleClient)
	channel := idleClient.channel
	if err = channel.Publish(
		p.exchange,
		p.routingKey,
		false,
		false,
		amqp.Publishing{
			Headers:         amqp.Table{},
			ContentType:     "text/plain",
			ContentEncoding: "",
			Body:            []byte(message),
			DeliveryMode:    amqp.Transient,
			Priority:        0,
		}); err != nil {
		idleClient.destroy = true
	}
	return err
}

func (p *Pool) incrFaileNum() {
	atomic.AddInt64(&p.failNum, 1)
}

func (p *Pool) resetFaileNum() {
	p.failNum = 0
}

func (p *Pool) makeNewConnection() error {
	connection, err := amqp.Dial(p.url)
	if err != nil {
		log.Errorf("dial to %s error:%+v", p.url, err)
		return err
	}
	channel, err := p.initConnection(connection)
	if err != nil {
		log.Errorf("connection init error:%s,%+v", p.url, err)
		connection.Close()
		return err
	}
	p.idle.PushBack(&IdleClient{
		connection: connection,
		channel:    channel,
		createTime: time.Now(),
		destroy:    false})
	for i := 0; i < DEFAULTCHANNEL-1; i++ {
		channel, err = connection.Channel()
		if err != nil {
			log.Errorf("channel get error:%s,%+v", p.url, err)
			connection.Close()
			return err
		}
		p.idle.PushBack(&IdleClient{
			connection: connection,
			channel:    channel,
			createTime: time.Now(),
			destroy:    false})
	}
	return nil
}

func (p *Pool) initConnection(connection *amqp.Connection) (*amqp.Channel, error) {
	channel, err := connection.Channel()
	if err != nil {
		log.Errorf("got channel error: %+v", err)
		return nil, err
	}
	if err = channel.ExchangeDeclare(
		p.exchange,   // name
		EXCHANGETYPE, // type
		true,         // durable
		false,        // auto-deleted
		false,        // internal
		false,        // noWait
		nil,          // arguments
	); err != nil {
		log.Errorf("exchange declare: %+v", err)
		return nil, err
	}
	if _, err = channel.QueueDeclare(
		p.routingKey,
		true,
		false,
		false,
		false,
		nil,
	); err != nil {
		log.Errorf("queue declare: %+v", err)
		return nil, err
	}
	if err = channel.QueueBind(
		p.routingKey,
		p.routingKey,
		p.exchange,
		false,
		nil,
	); err != nil {
		log.Errorf("queue bind error: %q", err)
		return nil, err
	}
	log.Infof("exchange declared, %q bind queue %q to exchange %q", p.url, p.routingKey, p.exchange)
	return channel, nil
}

func (p *Pool) Get() (*IdleClient, error) {
	p.lock.Lock()
	if p.idle.Len() == 0 && p.count >= p.maxConn {
		p.lock.Unlock()
		return nil, ErrOverMax
	}
	log.Infof("%s pool, before get, pool's idle len is:%d", p.GetPoolKey(), p.idle.Len())
	if p.idle.Len() == 0 {
		err := p.makeNewConnection()
		if err != nil {
			p.lock.Unlock()
			return nil, err
		}
		p.count += DEFAULTCHANNEL
		ele := p.idle.Front()
		client, ok := ele.Value.(*IdleClient)
		//xwj add begin 2019/1/17
		if !ok {
			log.Errorf("It's not ok for type IdleClient,ele.Value[%+v]", ele.Value)
			p.lock.Unlock()
			return nil, errors.New("It's not ok for type IdleClient")
		}
		//xwj add end 2019/1/17
		p.idle.Remove(ele)
		p.lock.Unlock()
		if !client.Check() {
			p.lock.Lock()
			if p.count > 0 {
				p.count -= 1
			}
			p.lock.Unlock()
			return nil, ErrSocketDisconnect
		}
		return client, nil
	} else {
		ele := p.idle.Front()
		idlec, ok := ele.Value.(*IdleClient)
		//xwj add begin 2019/1/17
		if !ok {
			log.Errorf("It's not ok for type IdleClient,ele.Value[%+v]", ele.Value)
			p.lock.Unlock()
			return nil, errors.New("It's not ok for type IdleClient")
		}
		//xwj add end 2019/1/17
		p.idle.Remove(ele)
		p.lock.Unlock()

		if !idlec.Check() {
			p.lock.Lock()
			if p.count > 0 {
				p.count -= 1
			}
			p.lock.Unlock()
			return nil, ErrSocketDisconnect
		}
		return idlec, nil
	}
}

func (p *Pool) Put(client *IdleClient) error {
	if client == nil {
		return ErrInvalidConn
	}
	log.Infof("%s pool, before put, pool's idle len is:%d", p.GetPoolKey(), p.idle.Len())
	p.lock.Lock()

	if p.count > p.maxConn {
		if p.count > 0 {
			p.count -= 1
		}
		p.lock.Unlock()

		err := p.Close(client)
		client = nil
		return err
	}

	if !client.Check() {
		if p.count > 0 {
			p.count -= 1
		}
		p.lock.Unlock()

		err := p.Close(client)
		client = nil
		return err
	}

	p.idle.PushBack(client)
	p.lock.Unlock()
	log.Infof("%s pool, after put, pool's idle len is:%d", p.GetPoolKey(), p.idle.Len())

	return nil
}

func (p *Pool) CheckTimeout() {
	log.Infof("before idle timeout check,%s pool's idle len:%d", p.GetPoolKey(), p.idle.Len())
	p.lock.Lock()
	for p.idle.Len() > p.initConn*DEFAULTCHANNEL {
		ele := p.idle.Back()
		if ele == nil {
			break
		}
		v, ok := ele.Value.(*IdleClient)
		//xwj add begin 2019/1/17
		if !ok {
			log.Errorf("It's not ok for type IdleClient,ele.Value[%+v]", ele.Value)
			p.lock.Unlock()
			break
		}
		//xwj add end 2019/1/17
		if v.createTime.Add(p.idleTimeout).After(nowFunc()) {
			break
		}

		p.idle.Remove(ele)
		p.lock.Unlock()
		p.Close(v) //close client connection
		p.lock.Lock()
		if p.count > 0 {
			p.count -= 1
		}
	}
	p.lock.Unlock()
	log.Infof("after idle timeout check,%s pool's idle len:%d", p.GetPoolKey(), p.idle.Len())

	return
}
func (p *Pool) Close(idleClient *IdleClient) error {
	p.lock.Lock()
	var n *list.Element
	for e := p.idle.Front(); e != nil; e = n {
		n = e.Next()
		cli, ok := e.Value.(*IdleClient)
		//xwj add begin 2019/1/17
		if !ok {
			log.Errorf("It's not ok for type IdleClient,e.Value[%+v]", e.Value)
			p.lock.Unlock()
			return errors.New("It's not ok for type IdleClient")
		}
		//xwj add end 2019/1/17
		if cli.connection == idleClient.connection {
			p.idle.Remove(e)
			if p.count > 0 {
				p.count -= 1
			}
		}
	}
	p.lock.Unlock()
	return idleClient.connection.Close()
}
func (p *Pool) InitPool() error {
	for i := 0; i < p.initConn; i++ {
		err := p.makeNewConnection()
		if err != nil {
			log.Errorf("%s pool init error: %+v", p.GetPoolKey(), err)
			return errors.New("pool init error!!")
		}
	}
	return nil
}

func (c *IdleClient) Check() bool {
	if c.channel == nil || c.destroy {
		return false
	}
	return true
}
func (p *Pool) GetPoolKey() string {
	return p.url
}
func (p *Pool) GetIdleCount() int {
	return p.idle.Len()
}

func (p *Pool) GetConnCount() int {
	return p.count
}

func (pm *PoolManager) StartIdleConnClearTask() {
	for {
		for poolEle := pm.poolList.Front(); poolEle != nil; poolEle = poolEle.Next() {
			p, ok := poolEle.Value.(*Pool)
			//xwj add begin 2019/1/17
			if !ok {
				log.Errorf("It's not ok for type Pool,ele.Value[%+v]", poolEle.Value)
				break
			}
			//xwj add end 2019/1/17
			p.CheckTimeout()
		}

		time.Sleep(IDLECHECKINTERVAL * time.Second)
	}
}
